package cn.dmp.tools

import ch.hsr.geohash.GeoHash
import cn.dmp.utils.JedisPools
import org.apache.commons.lang.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * 将日志字段的经度和纬度提取出来，将其转换成对应的GeoHashCode, 以及对应的商圈信息
  */
object LatLong2Bussiness {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf()
          .setMaster("local[*]")
          .setAppName("商圈知识库的建立")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(sparkConf)
        val sQLContext = new SQLContext(sc)

        val dataFrame = sQLContext.read.parquet("F:\\Temp\\parquet1")


        // 提取出数据中的经度和纬度 -> redis
        dataFrame.select("lat", "long").distinct()
          .where("lat>=3.86 and lat<=53.55 and long>=73.66 and long<=135.05")
          .map(row => {
              val lat = row.getAs[String]("lat")
              val long = row.getAs[String]("long")

              // GeoHashCode
              val geohashCode = GeoHash.withCharacterPrecision(lat.toDouble, long.toDouble, 8).toBase32

              // 商圈信息 -> Baidu LBS
              val businessTag = BaiduLBS.businessInfo(lat, long)
              (geohashCode, businessTag)
          })
          .foreachPartition{
              case itr => {
                  val jedis = JedisPools.getJedis
                  itr.foreach(tp => {
                      if (StringUtils.isNotEmpty(tp._2))
                          jedis.hset("busdict", tp._1, tp._2)

                      // jedis.expire("A", 60 *5) 设置key 有效期为5分钟

                  })
                  jedis.close()
              }
          }

        sc.stop()

    }

}
